package com.atguigu.gmall.realtime.common.util;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.common.constant.Constant;
import com.google.common.base.CaseFormat;
import org.apache.commons.beanutils.BeanUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.util.List;
import java.util.Set;

/**
 * @author Felix
 * @date 2024/6/19
 * 操作HBase的工具类
 */
public class HBaseUtil {
    //获取HBase连接
    public static Connection getHBaseConnection() {
        try {
            Configuration conf = new Configuration();
            conf.set("hbase.zookeeper.quorum", "hadoop102");
            //conf.set("hbase.zookeeper.property.clientPort", "2181");
            Connection connection = ConnectionFactory.createConnection(conf);
            return connection;
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    //关闭HBase连接
    public static void closeHBaseConnection(Connection hbaseConn) {
        if (hbaseConn != null && !hbaseConn.isClosed()) {
            try {
                hbaseConn.close();
            } catch (IOException e) {
                throw new RuntimeException(e);
            }
        }
    }

    //获取异步操作HBase连接对象
    public static AsyncConnection getAsyncHBaseConnection() {
        try {
            Configuration conf = new Configuration();
            conf.set("hbase.zookeeper.quorum", "hadoop102");
            //conf.set("hbase.zookeeper.property.clientPort", "2181");
            AsyncConnection asyncConn = ConnectionFactory.createAsyncConnection(conf).get();
            return asyncConn;
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

    //关闭异步操作HBase连接对象
    public static void closeAsyncHBaseConnection(AsyncConnection asyncHBaseConn) {
        if (asyncHBaseConn != null && !asyncHBaseConn.isClosed()) {
            try {
                asyncHBaseConn.close();
            } catch (IOException e) {
                throw new RuntimeException(e);
            }
        }
    }

    /**
     * 在HBase中建表
     *
     * @param hbaseConn 连接
     * @param nameSpace 表空间
     * @param tableName 表名
     * @param families  列族
     */
    public static void createHBaseTable(Connection hbaseConn, String nameSpace, String tableName, String... families) {
        //判断是否传递了列族
        if (families.length < 1) {
            System.out.println("建表必须提供列族");
            return;
        }
        try (Admin admin = hbaseConn.getAdmin()) {
            //判断表是否存在
            TableName tableNameObj = TableName.valueOf(nameSpace, tableName);
            if (admin.tableExists(tableNameObj)) {
                System.out.println("要创建的" + nameSpace + "下的" + tableName + "表已存在");
                return;
            }
            TableDescriptorBuilder tableDescriptorBuilder = TableDescriptorBuilder.newBuilder(tableNameObj);
            //指定列族
            for (String family : families) {
                ColumnFamilyDescriptor columnFamilyDescriptor
                        = ColumnFamilyDescriptorBuilder.newBuilder(Bytes.toBytes(family)).build();
                tableDescriptorBuilder.setColumnFamily(columnFamilyDescriptor);
            }
            admin.createTable(tableDescriptorBuilder.build());
            System.out.println("在HBase中创建了" + nameSpace + "下的" + tableName + "表");
        } catch (IOException e) {
            throw new RuntimeException(e);
        }

    }

    /**
     * 从HBase中删除表
     *
     * @param hbaseConn 连接对象
     * @param nameSpace 表空间
     * @param tableName 表名
     */
    public static void dropHBaseTable(Connection hbaseConn, String nameSpace, String tableName) {
        try (Admin admin = hbaseConn.getAdmin()) {
            TableName tableNameObj = TableName.valueOf(nameSpace, tableName);
            if (!admin.tableExists(tableNameObj)) {
                System.out.println("要删除的" + nameSpace + "下的" + tableName + "表不存在");
                return;
            }
            admin.disableTable(tableNameObj);
            admin.deleteTable(tableNameObj);
            System.out.println("从HBase中删除了" + nameSpace + "下的" + tableName + "表");

        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    //从HBase表中删除记录
    public static void delRow(Connection hbaseConn, String nameSpace, String tableName, String rowKey) {
        TableName tableNameObj = TableName.valueOf(nameSpace, tableName);
        try (Table table = hbaseConn.getTable(tableNameObj)) {
            Delete delete = new Delete(Bytes.toBytes(rowKey));
            table.delete(delete);
            System.out.println("从" + nameSpace + "的" + tableName + "表中删除" + rowKey + "数据成功");
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    /**
     * 向HBase表中put数据
     *
     * @param hbaseConn 连接对象
     * @param nameSpace 表空间
     * @param tableName 表名
     * @param rowKey    rowkey
     * @param family    列族
     * @param jsonObj   上游传递过来的dataJson
     */

    public static void putRow(Connection hbaseConn, String nameSpace, String tableName, String rowKey, String family, JSONObject jsonObj) {
        // jsonObj : {"tm_name":"Redmi123","id":1}
        TableName tableNameObj = TableName.valueOf(nameSpace, tableName);
        try (Table table = hbaseConn.getTable(tableNameObj)) {
            Put put = new Put(Bytes.toBytes(rowKey));

            Set<String> columns = jsonObj.keySet();
            for (String column : columns) {
                String value = jsonObj.getString(column);
                if (StringUtils.isNotEmpty(value)) {
                    put.addColumn(Bytes.toBytes(family), Bytes.toBytes(column), Bytes.toBytes(value));
                }
            }
            table.put(put);
            System.out.println("向" + nameSpace + "的" + tableName + "表中put数据" + rowKey + "成功");
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    //根据rowkey到Hbase表中获取数据
    public static <T>T getRow(Connection hbaseConn,String nameSpace,String tableName,String rowkey,Class<T> clz, boolean... isUnderLineToCamel){
        boolean defaultIsUToC = false;  // 默认不执行下划线转驼峰

        if (isUnderLineToCamel.length > 0) {
            defaultIsUToC = isUnderLineToCamel[0];
        }

        TableName tableNameObj = TableName.valueOf(nameSpace, tableName);
        try (Table table = hbaseConn.getTable(tableNameObj)){
            Get get = new Get(Bytes.toBytes(rowkey));
            Result result = table.get(get);
            List<Cell> cells = result.listCells();
            if(cells != null && cells.size() > 0){
                //定义一个对象，用于封装查询出来的维度数据
                T obj = clz.newInstance();
                for (Cell cell : cells) {
                    String columnName = Bytes.toString(CellUtil.cloneQualifier(cell));
                    String columnValue = Bytes.toString(CellUtil.cloneValue(cell));
                    if(defaultIsUToC){
                        columnName = CaseFormat.LOWER_UNDERSCORE.to(CaseFormat.LOWER_CAMEL,columnName);
                    }
                    BeanUtils.setProperty(obj,columnName,columnValue);
                }
                return obj;
            }

        } catch (Exception e) {
            throw new RuntimeException(e);
        }

        return null;
    }

    //以异步的方式从HBase中获取维度数据
    public static JSONObject readDimAsync(AsyncConnection asyncHbaseConn,String nameSpace,String tableName,String rowkey){
        TableName tableNameObj = TableName.valueOf(nameSpace, tableName);
        AsyncTable<AdvancedScanResultConsumer> asyncTable = asyncHbaseConn.getTable(tableNameObj);
        Get get = new Get(Bytes.toBytes(rowkey));
        try {
            Result result = asyncTable.get(get).get();
            List<Cell> cells = result.listCells();
            if(cells != null && cells.size() > 0){
                //定义一个json对象，用于封装查询结果
                JSONObject jsonObj = new JSONObject();
                for (Cell cell : cells) {
                    String columnName = Bytes.toString(CellUtil.cloneQualifier(cell));
                    String columnValue = Bytes.toString(CellUtil.cloneValue(cell));
                    jsonObj.put(columnName,columnValue);
                }
                return jsonObj;
            }
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
        return null;
    }

    public static void main(String[] args) {
        Connection hBaseConn = getHBaseConnection();
        System.out.println(getRow(hBaseConn, Constant.HBASE_NAMESPACE, "dim_base_trademark", "1", JSONObject.class));
        closeHBaseConnection(hBaseConn);
    }

}
